package com.bw.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.bw.bean.VisitorStats;
import com.bw.utils.ClickHouseUtil;
import com.bw.utils.DateTimeUtil;
import com.bw.utils.MyKafkaUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;
import java.time.Duration;
import java.util.Date;

// 数据源：App/Web/小程序--->nginx--->SpringBoot--->Kafka(ods)--->Flink--->Kafka(DWD)---->Flink---->Kafka(DWM)--->Flink--->CK(dws)

// 程序员： jar  ----nginx---springboot--->kafka---->BaseLogAPP--->Kafka(DWD)--->UniqueVisitAPP/UserJumpDetailApp-->Kafka(DWM)---VisitorStatsApp
public class VisitorStatsApp {
    public static void main(String[] args) throws Exception {
        // 1. 获取执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        //        开启CK，创建文件夹
//        env.setStateBackend(new FsStateBackend("hdfs://hadoop102:9820/gmall-flink/ck"));
//         每隔5秒保存一次
//        env.enableCheckpointing(5000L);
//        精准消费一次
//        env.getCheckpointConfig().setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
//        配置超时时间
//        env.getCheckpointConfig().setCheckpointTimeout(100000L);
//        同时最大可以运行几个ck
//        env.getCheckpointConfig().setMaxConcurrentCheckpoints(2);
//        上一个ck的尾部到下一个ck的头之间间隔
//        env.getCheckpointConfig().setMinPauseBetweenCheckpoints(3000L);
        // env.getCheckpointConfig().enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 2. 读取Kafka主题 dwd_page_log dwm_unique_visit dwm_user_jump_detail
        String uniqueVisitSourceTopic = "dwm_unique_visit";
        String userJumpDetailSourceTopic = "dwm_user_jump_detail";
        String pageViewSourceTopic = "dwd_page_log";
        String groupId = "visitor_stats_app_2105b_1";
        DataStreamSource<String> uvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(uniqueVisitSourceTopic, groupId));
        DataStreamSource<String> ujDS = env.addSource(MyKafkaUtil.getKafkaConsumer(userJumpDetailSourceTopic, groupId));
        DataStreamSource<String> pvDS = env.addSource(MyKafkaUtil.getKafkaConsumer(pageViewSourceTopic, groupId));
        // 3. 每个流转换成相应的JavaBean对象  3 个流
        SingleOutputStreamOperator<VisitorStats> uvVisitorStatsDS = uvDS.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                // 转成jsonobject
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return new VisitorStats("", "", common.getString("vc"),
                        common.getString("ch"), common.getString("ar"),
                        common.getString("is_new"), 1L, 0L, 0L, 0L, 0L,
                        jsonObject.getLong("ts"));
            }
        });
        SingleOutputStreamOperator<VisitorStats> ujVisitorStatsDS = ujDS.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                return new VisitorStats("", "", common.getString("vc"),
                        common.getString("ch"), common.getString("ar"),
                        common.getString("is_new"), 0L, 0L, 0L, 1L, 0L,
                        jsonObject.getLong("ts"));
            }
        });
        SingleOutputStreamOperator<VisitorStats> pvVisitorStatsDS = pvDS.map(new MapFunction<String, VisitorStats>() {
            @Override
            public VisitorStats map(String value) throws Exception {
                JSONObject jsonObject = JSON.parseObject(value);
                JSONObject common = jsonObject.getJSONObject("common");
                JSONObject page = jsonObject.getJSONObject("page");
                String lastPageId = page.getString("last_page_id");
                long sv = 0L;
                if (lastPageId == null || lastPageId.length() <= 0) {
                    sv = 1L;
                }
                return new VisitorStats("", "", common.getString("vc"),
                        common.getString("ch"), common.getString("ar"),
                        common.getString("is_new"), 0L, 1L, sv, 0L, page.getLong("during_time"),
                        jsonObject.getLong("ts"));
            }
        });

        // 4. 把多个流union就
        DataStream<VisitorStats> unionDS = uvVisitorStatsDS.union(ujVisitorStatsDS, pvVisitorStatsDS);
        // 5. 生成水印
        SingleOutputStreamOperator<VisitorStats> unionWithWmDs = unionDS.assignTimestampsAndWatermarks(WatermarkStrategy.<VisitorStats>forBoundedOutOfOrderness(Duration.ofSeconds(12)).withTimestampAssigner(new SerializableTimestampAssigner<VisitorStats>() {
            @Override
            public long extractTimestamp(VisitorStats visitorStats, long l) {
                return visitorStats.getTs();
            }
        }));
        // 6. 根据4个维度Key By
        KeyedStream<VisitorStats, Tuple4<String, String, String, String>> keyByDS = unionWithWmDs.keyBy(new KeySelector<VisitorStats, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(VisitorStats visitorStats) throws Exception {
                return Tuple4.of(visitorStats.getVc(), visitorStats.getCh(), visitorStats.getAr(), visitorStats.getIs_new());
            }
        });
        // 7、开窗聚合 滚动窗口10s
        SingleOutputStreamOperator<VisitorStats> reduceDS = keyByDS.window(TumblingEventTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<VisitorStats>() {

            @Override
            public VisitorStats reduce(VisitorStats v1, VisitorStats v2) throws Exception {
//                return new VisitorStats(
//                       v1.getStt(),
//                       v1.getEdt(),
//                       v1.getVc(),
//                        v1.getCh(),
//                        v1.getAr(),
//                        v1.getIs_new(),
//                        v1.getUv_ct()+v2.getUv_ct(),
//                );
                // 滚动窗口这么写没问题，滑动有问题
                v1.setPv_ct(v1.getPv_ct() + v2.getPv_ct());
                v1.setUv_ct(v1.getUv_ct() + v2.getUv_ct());
                v1.setUj_ct(v1.getUj_ct() + v2.getUj_ct());
                v1.setSv_ct(v1.getSv_ct() + v2.getSv_ct());
                v1.setDur_sum(v1.getDur_sum() + v2.getDur_sum());
                return v1;
            }
            // ctrl+p 查看参数
        }, new WindowFunction<VisitorStats, VisitorStats, Tuple4<String, String, String, String>, TimeWindow>() {
            @Override
            public void apply(Tuple4<String, String, String, String> stringStringStringStringTuple4, TimeWindow timeWindow, Iterable<VisitorStats> iterable, Collector<VisitorStats> collector) throws Exception {
                long start = timeWindow.getStart();
                long end = timeWindow.getEnd();
                VisitorStats visitorStats = iterable.iterator().next();
                visitorStats.setStt(DateTimeUtil.toYMDhms(new Date(start)));
                visitorStats.setEdt(DateTimeUtil.toYMDhms(new Date(end)));
                // 收集
                collector.collect(visitorStats);
            }
        });
        // 8、写入CK(clickhouse)
        reduceDS.print();

        reduceDS.addSink(ClickHouseUtil.sink("insert into table product_stats_2105b values(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)"));
        // 9. 启动任务
        env.execute("VisitorStatsApp");

    }
}
